Fixes #24788
Fixes #24509

Motivation

The test is flaky.

Reproduce the issue:
Add Thread.sleep after sendAsync to avoid batching sends.

 for (int i = 0; i < 50; i++) {
       producer.newMessage(transaction).value(i).sendAsync();
       Thread.sleep(1);
}

The issue occurs because the in this test, the TransactionBufferTestImpl class extends TopicTransactionBuffer and declares a field with the same name publishFuture.
Therefore, in the appendBufferToTxn method, getPublishFuture() retrieves the publishFuture from the subclass, while the assignment publishFuture = future modifies the publishFuture belonging to the parent class.

 CompletableFuture<Position> future = getPublishFuture().thenCompose(ignore -> {
            ......
        }).whenComplete(((position, throwable) -> buffer.release()));
publishFuture = future;

Modifications

Use ‘setPublishFuture(future)’ to replace ‘publishFuture = future’

Verifying this change

• Make sure that the change passes the CI checks.

This change is already covered by existing tests.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: Dream95#1